package com.atguigu.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp;
import com.atguigu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/4 12:13
 */
public class DWDLogApp extends BaseApp {
    public static void main(String[] args) {
        new DWDLogApp().init(1, "DWDLogApp", "ods_log");
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       DataStreamSource<String> sourceStream) {
        setWebUi(env, 1999);
        
        // 1. 识别新老客户: 在日志数据中添加一个字段用来表示是否新老客户
        SingleOutputStreamOperator<JSONObject> validateFlagDS = distinguishNewOrOld(sourceStream);
        // 2. 流的拆分: 利用侧输出流  返回结果: 页面日志 启动日志 曝光日志
        Tuple3<DataStream<String>, DataStream<String>, DataStream<String>> streams = splitStream(validateFlagDS);
        // 3. 分别写入到不同的Topic中
        sendToKafka(streams);
        
    }
    
    private void sendToKafka(Tuple3<DataStream<String>, DataStream<String>, DataStream<String>> streams) {
        streams.f0.addSink(MyKafkaUtil.getKafkaSink("dwd_page_log"));
        streams.f1.addSink(MyKafkaUtil.getKafkaSink("dwd_start_log"));
        streams.f2.addSink(MyKafkaUtil.getKafkaSink("dwd_display_log"));
    }
    
    private SingleOutputStreamOperator<JSONObject> distinguishNewOrOld(DataStreamSource<String> sourceStream) {
        SingleOutputStreamOperator<JSONObject> result = sourceStream
            // 数据解析成JSONObject对象
            .map(JSON::parseObject).setParallelism(1)
            .assignTimestampsAndWatermarks(  // 添加水印和事件时间
                                             WatermarkStrategy
                                                 .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                                                 .withTimestampAssigner((element, recordTimestamp) -> element.getLong("ts"))
            )
            .keyBy(jsonData -> jsonData.getJSONObject("common").getString("mid"))
            // 按照mid进行分组
            .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 添加滚动窗口
            .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
                
                private ValueState<Long> firstVisitState;  // 存储是是首次访问的时间戳
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    firstVisitState = getRuntimeContext()
                        .getState(new ValueStateDescriptor<>("firstVisitState", Long.class));
                }
                
                @Override
                public void process(String mid,
                                    Context context,
                                    Iterable<JSONObject> elements,
                                    Collector<JSONObject> out) throws Exception {
                    if (firstVisitState.value() == null) {
                        // 1.  如果是这个 mid 的首个窗口, 则把时间戳最小的设置首次访问, 其他均为非首次访问
                        ArrayList<JSONObject> list = new ArrayList<>();
                        for (JSONObject element : elements) {
                            list.add(element);
                        }
                        
                        list.sort(Comparator.comparing(o -> o.getLong("ts")));
                        for (int i = 0; i < list.size(); i++) {
                            if (i == 0) {
                                list.get(i).getJSONObject("common").put("is_new", "1");
                                firstVisitState.update(list.get(i).getLong("ts"));
                            } else {
                                list.get(i).getJSONObject("common").put("is_new", "0");
                            }
                            out.collect(list.get(i));
                        }
                    } else {
                        // 2. 如果不是这个mid的首个窗口, 则所有均为非首次访问
                        elements.forEach(data -> {
                            data.getJSONObject("common").put("is_new", "0");
                            out.collect(data);
                        });
                    }
                }
            });
        return result;
    }
    
    private Tuple3<DataStream<String>, DataStream<String>, DataStream<String>> splitStream(SingleOutputStreamOperator<JSONObject> stream) {
        OutputTag<String> startTag = new OutputTag<String>("startStream") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayStream") {};
        
        SingleOutputStreamOperator<String> pageStream = stream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value,
                                       Context ctx,
                                       Collector<String> out) throws Exception {
                
                // 1. 获取启动数据
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    // 2. 如果是启动日志, 则把数据放入启动侧输出流
                    ctx.output(startTag, value.toJSONString());
                } else {
                    // 3. 如果非启动日志, 则为页面日志或者曝光数据或者既是页面又是曝光
                    // 3.1 是否为页面日志
                    final JSONObject page = value.getJSONObject("page");
                    if (page != null) {
                        out.collect(value.toJSONString());
                    }
                    // 3.2 获取曝光信息, 每个曝光信息为流中的一条数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            // 给display补充 pageId 信息
                            String pageId = value.getJSONObject("page").getString("page_id");
                            display.put("page_id", pageId);
                            ctx.output(displayTag, display.toJSONString());
                        }
                    }
                }
            }
        });
        
        DataStream<String> startStream = pageStream.getSideOutput(startTag);
        DataStream<String> displayStream = pageStream.getSideOutput(displayTag);
        
        return Tuple3.of(pageStream, startStream, displayStream);
    }
}
